package com.shujia.state

import java.lang

import org.apache.flink.api.common.functions.{RichMapFunction, RuntimeContext}
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object Demo2ListState {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val student: DataStream[String] = env.socketTextStream("master",8888)

    /**
      * 实时计算班级平均年龄
      */

    val kvDS: DataStream[(String, Int)] = student.map(line =>{
      val split: Array[String] = line.split(",")
      (split(4),split(2).toInt)
    })

    kvDS
      .keyBy(_._1)
      .map(new MyRichMapFunction)
      .print()

    env.execute()

  }

}

class MyRichMapFunction extends RichMapFunction[(String,Int),(String,Double)]{


  var listState: ListState[Int] = _

  override def open(parameters: Configuration): Unit = {

    //获取flink环境，定义状态
    val context: RuntimeContext = getRuntimeContext

    /**
      * ListState；为每一个key保存一个集合状态
      */

    val stateDesc = new ListStateDescriptor[Int]("avg",classOf[Int])

    listState = context.getListState(stateDesc)

  }

  override def map(value: (String, Int)): (String, Double) = {

    //往集合中增加元素
    listState.add(value._2)

    //获取所有年龄
    val iterable: lang.Iterable[Int] = listState.get()
    import scala.collection.JavaConversions._
    val ages: List[Int] = iterable.toList

    //计算平均年龄
    val avgAge: Double = ages.sum.toDouble / ages.length

    //返回数据
    (value._1,avgAge)

  }
}
